﻿using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using Common.Utilities;
using Common.Utilities.Dispatcher;
using System.Diagnostics;

namespace Workflows.Components.TextMining.Collapse
{
    /// <summary>
    /// 
    /// </summary>
    public class EntityLinkWorker:IWorker 
    {
        #region Implementation of IWorker

        public event WorkerStarted WorkerStartedEvent;
        public event WorkerFinished WorkerFinishedEvent;
        public event WorkerHasError WorkerErredEvent;
        public event WorkerCancelled WorkerCancelledEvent;
        public event WorkerStatusChanged WorkerStatusChangedEvent;

        private Thread workerThread;
        private EntityLinkInstruction instruction;
        private EntityLinkOutput output;
        private RunContext _Context;
        private Guid _TicketID;
        private string _WorkflowName;
        private string _ComponentName;

        /// <summary>
        /// 
        /// </summary>
        public string ComponentName
        {
            get { return _ComponentName; }
        }

        /// <summary>
        /// 
        /// </summary>
        public string WorkflowName
        {
            get { return _WorkflowName; }
        }

        /// <summary></summary>
        public Thread WorkerThread
        {
            get { return workerThread; }
        }
        /// <summary></summary>
        public InstructionBase Instruction
        {
            get { return instruction; }
        }
        /// <summary></summary>
        public ResultBase Output
        {
            get { return output; }
        }
        /// <summary></summary>
        public Type ComponentType
        {
            get { return typeof(EntityLinkComponent); }
        }
        /// <summary></summary>
        public RunContext Context
        {
            get { return _Context; }
            set { _Context = value; }
        }
        /// <summary></summary>
        public void SetInstruction(Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._TicketID = ticketGuid;
            this.instruction = (EntityLinkInstruction)request;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }
        /// <summary></summary>
        public void SetInstruction(RunContext context, Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._TicketID = ticketGuid;
            this.instruction = (EntityLinkInstruction)request;
            this._Context = context;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }
        /// <summary></summary>
        public void DoWork()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._TicketID, this, this.GetType(), this.instruction);
            }
            this.workerThread = Thread.CurrentThread;
            this.LinkEntities();
        }
        /// <summary></summary>
        public void DoWorkAsync()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._TicketID, this, this.GetType(), this.instruction);
            }
            this.workerThread = new Thread(new ThreadStart(this.LinkEntities));
            this.workerThread.Start();
        }
        /// <summary></summary>
        public void CancelWork()
        {
            if (this.workerThread != null && this.workerThread.IsAlive)
            {
                this.workerThread.Abort();
                this.workerThread.Join();

                if (this.WorkerCancelledEvent != null)
                {
                    this.WorkerCancelledEvent(this._TicketID, this.GetType(), this.instruction);
                }
            }
        }

        #endregion

        #region work
        private void LinkEntities()
        {
            try
            {
                int entity1Count = 0;
                int entity2Count = 0;
                int entityLinked = 0;

                Dictionary<string,IAttributeComparer> comparers=new Dictionary<string, IAttributeComparer>();
                Assembly assembly = this.GetType().Assembly;
                foreach(string fieldName in this.instruction.FieldComparers.Keys)
                {
                    string comparerTypeName = this.instruction.FieldComparers[fieldName];
                    Type type = assembly.GetType(comparerTypeName);
                    if(type.GetInterface(typeof(IAttributeComparer).FullName)!=null)
                    {
                        IAttributeComparer comparer = (IAttributeComparer) Activator.CreateInstance(type);
                        comparers.Add(fieldName, comparer);
                    }
                }

                Dictionary<object, Dictionary<object, double>> key2_key1_scores = 
                    new Dictionary<object, Dictionary<object, double>>();
                foreach(string descField1 in this.instruction.DescFields1)
                {
                    string field1 = descField1;
                    string field2 = this.instruction.Field1_Field2_Mappings[field1];
                    Dictionary<object, object> keyValues1 = this.GetFieldValues(
                        this.instruction.ConnStr1, this.instruction.SchemaName1,
                        this.instruction.TableName1, this.instruction.Sql1,
                        this.instruction.PKField1, field1);
                    if (keyValues1.Count > entity1Count)
                        entity1Count = keyValues1.Count;
                    Dictionary<object, object> keyValues2 = this.GetFieldValues(
                        this.instruction.ConnStr2, this.instruction.SchemaName2,
                        this.instruction.TableName2, this.instruction.Sql2,
                        this.instruction.PKField2, field2);
                    if (keyValues2.Count > entity2Count)
                        entity2Count = keyValues2.Count;
                    IAttributeComparer comparer = comparers[field1];
                    double weight = this.instruction.FieldWeights[field1];
                    double penalty = this.instruction.FieldPenalties[field1];
                    this.CompareField(field1, comparer, weight, penalty,
                                      keyValues1, keyValues2, ref key2_key1_scores);
                }
                List<object> keys2=new List<object>();
                keys2.AddRange(key2_key1_scores.Keys);
                foreach (object key2 in keys2)
                {
                    Dictionary<object, double> toEntity1Scores = key2_key1_scores[key2];
                    Dictionary<object,double> copy=new Dictionary<object, double>();
                    foreach(object key1 in toEntity1Scores.Keys)
                    {
                        if(toEntity1Scores[key1]>=this.instruction.AssocThreshold)
                            copy.Add(key1, toEntity1Scores[key1]);
                    }
                    key2_key1_scores[key2] = copy;
                }
                foreach(string assocField1 in this.instruction.AssocFields1)
                {
                    string field1 = assocField1;
                    string field2 = this.instruction.Field1_Field2_Mappings[field1];
                    Dictionary<object, object> keyValues1 = this.GetFieldValues(
                        this.instruction.ConnStr1, this.instruction.SchemaName1,
                        this.instruction.TableName1, this.instruction.Sql1,
                        this.instruction.PKField1, field1);
                    if (keyValues1.Count > entity1Count)
                        entity1Count = keyValues1.Count;
                    Dictionary<object, object> keyValues2 = this.GetFieldValues(
                        this.instruction.ConnStr2, this.instruction.SchemaName2,
                        this.instruction.TableName2, this.instruction.Sql2,
                        this.instruction.PKField2, field2);
                    if (keyValues2.Count > entity2Count)
                        entity2Count = keyValues2.Count;
                    IAttributeComparer comparer = comparers[field1];
                    double weight = this.instruction.FieldWeights[field1];
                    double penalty = this.instruction.FieldPenalties[field1];
                    this.CompareField(field1, comparer, weight, penalty,
                                      keyValues1, keyValues2, ref key2_key1_scores);
                }
                
                Dictionary<object, Dictionary<object, double>> linkageFromEntity2ToEntity1 =
                    this.MergeLinkages(key2_key1_scores, this.instruction.LinkThreshold);

                this.SaveLinkOutput(linkageFromEntity2ToEntity1);
                
                entityLinked = linkageFromEntity2ToEntity1.Count;
                this.output=new EntityLinkOutput(this.instruction.OwnerComponentID,this._TicketID,
                    entity1Count,entity2Count,entityLinked);
                if (this.WorkerFinishedEvent != null)
                    this.WorkerFinishedEvent(this._TicketID, this.GetType(), this.instruction, this.output);
            }
            catch(Exception ex)
            {
                if (this.WorkerErredEvent != null)
                    this.WorkerErredEvent(this._TicketID, this.GetType(), this.instruction, ex.ToString(), true);
            }
        }

        private void SaveLinkOutput(Dictionary<object, Dictionary<object, double>> key2_key1_scores)
        {
            TableColumn idCol1 = new TableColumn();
            if (string.IsNullOrEmpty(this.instruction.Sql1))
                idCol1 = DbUtil.GetTableColumn(DataProviderType.MSSQL, this.instruction.ConnStr1, this.instruction.SchemaName1, this.instruction.TableName1, this.instruction.PKField1);
            else
                idCol1 = DbUtil.GetTableColumn(DataProviderType.MSSQL, this.instruction.ConnStr1, this.instruction.Sql1, this.instruction.PKField1);
            idCol1.ColumnName = this.instruction.EntityIDField1;
            TableColumn idCol2 = new TableColumn();
            if (string.IsNullOrEmpty(this.instruction.Sql1))
                idCol2 = DbUtil.GetTableColumn(DataProviderType.MSSQL, this.instruction.ConnStr2, this.instruction.SchemaName2, this.instruction.TableName2, this.instruction.PKField2);
            else
                idCol2 = DbUtil.GetTableColumn(DataProviderType.MSSQL, this.instruction.ConnStr2, this.instruction.Sql2, this.instruction.PKField2);
            idCol2.ColumnName = this.instruction.EntityIDField2;
            TableColumn scoreCol = new TableColumn();
            scoreCol.ColumnName = "Score";
            scoreCol.DbType = SqlDbType.Float;
            DbUtil.NewTable(DataProviderType.MSSQL, this.instruction.ConnStr1,
                this.instruction.SchemaName1, this.instruction.OutputTableName,
                new TableColumn[] { idCol1, idCol2, scoreCol });
            string sp_Name = string.Format("bulkInsert_{0}", this.instruction.OutputTableName);
            string sp_Body = StoredProcUtil.CreateStoredProcedureStatementForInsert(
                DataProviderType.MSSQL, sp_Name, this.instruction.SchemaName1, 
                this.instruction.OutputTableName,
                new TableColumn[] { idCol1, idCol2, scoreCol });
            DbUtil.NewStoredProc(this.instruction.ConnStr1, sp_Name, sp_Body);
            SqlConnection conn = new SqlConnection(this.instruction.ConnStr1);
            try
            {
                conn.Open();
                SqlCommand cmd = conn.CreateCommand();
                cmd.CommandType = CommandType.StoredProcedure;
                cmd.CommandText = sp_Name;
                int processed = 0;
                int total = key2_key1_scores.Count;
                foreach (object key2 in key2_key1_scores.Keys)
                {
                    if(this.WorkerStatusChangedEvent !=null && processed % 100==0)
                    {
                        string msg = string.Format("Saving linkages {0}/{1}: processed {2} of {3} records",
                                                   this.instruction.TableName1, this.instruction.TableName2, processed, total);
                        int pct = processed*100/total;
                        this.WorkerStatusChangedEvent(this._TicketID, this.GetType(), this.instruction,
                                                      this._WorkflowName, this._ComponentName, msg, pct);
                    }
                    foreach (object key1 in key2_key1_scores[key2].Keys)
                    {
                        double score = key2_key1_scores[key2][key1];
                        if (score >= this.instruction.LinkThreshold)
                        {
                            cmd.Parameters.Clear();
                            cmd.Parameters.Add(new SqlParameter("@" + idCol1.ColumnName, key1));
                            cmd.Parameters.Add(new SqlParameter("@" + idCol2.ColumnName, key2));
                            cmd.Parameters.Add(new SqlParameter("@" + scoreCol.ColumnName, score));
                            cmd.ExecuteNonQuery();
                        }
                    }
                    processed++;
                }
            }
            catch (Exception ex)
            {
                if(this.WorkerErredEvent !=null)
                    this.WorkerErredEvent(this._TicketID,this.GetType(),this.instruction,ex.ToString(),true);
            }
            finally
            {
                conn.Close();
            }
        }

        private Dictionary<object, object> GetFieldValues(
            string connStr, string schemaName, string tableName,
            string sql, string pkField, string fieldName)
        {
            Dictionary<object,object> keyVals=new Dictionary<object, object>();
            string selectSql = string.Empty;
            if (string.IsNullOrEmpty(sql))
                selectSql = string.Format(
                    "select [{0}], [{1}] from [{2}].[{3}]",
                    pkField, fieldName, schemaName, tableName);
            else
                selectSql = string.Format(
                    "select T1.{0}, T1.{1} from ({2}) T1",
                    pkField, fieldName, sql);
            DataTable dt = DbUtil.GetTableDataBySQL(connStr, "Output", selectSql);
            foreach(DataRow dr in dt.Rows)
            {
                object key = dr[pkField];
                object value = dr[fieldName];
                keyVals.Add(key,value);
            }
            return keyVals;
        }

        private void CompareField(
            string fieldName,
            IAttributeComparer comparer, 
            double weight, double penalty,
            Dictionary<object, object> keyValues1, 
            Dictionary<object, object> keyValues2, 
            ref Dictionary<object, Dictionary<object, double>> key2_key1_scores)
        {
            int processed = 0;
            int total = keyValues2.Count;
            foreach(object key2 in keyValues2.Keys)
            {
                if(this.WorkerStatusChangedEvent !=null && processed % 100==0)
                {
                    string msg = string.Format("Indexing {0}/{1} field {2}, processed {3} of {4} records",
                                               this.instruction.TableName1, this.instruction.TableName2,
                                               fieldName, processed, total);
                    int pct = processed*100/total;
                    this.WorkerStatusChangedEvent(this._TicketID, this.GetType(), this.instruction, this._WorkflowName,
                                                  this._ComponentName, msg, pct);
                }

                Dictionary<object, double> toEntity1Scores = new Dictionary<object, double>();
                if (key2_key1_scores.ContainsKey(key2))
                    toEntity1Scores = key2_key1_scores[key2];
                object val2 = keyValues2[key2];
                foreach(object key1 in keyValues1.Keys)
                {
                    double score = 0;
                    object val1 = keyValues1[key1];
                    double matchScore = comparer.Compare(val1, val2, penalty, null, null);
                    if(matchScore>0)
                        matchScore = matchScore*weight;
                    if (toEntity1Scores.ContainsKey(key1))
                        score = toEntity1Scores[key1];
                    score = score + matchScore;
                    if(toEntity1Scores.ContainsKey(key1))
                        toEntity1Scores[key1] = score;
                    else 
                        toEntity1Scores.Add(key1, score);
                }
                if(key2_key1_scores.ContainsKey(key2))
                    key2_key1_scores[key2] = toEntity1Scores;
                else 
                    key2_key1_scores.Add(key2, toEntity1Scores);

                processed++;
            }
        }

        private Dictionary<object, Dictionary<object, double>> MergeLinkages(
            Dictionary<object, Dictionary<object, double>> key2_key1_scores, double linkThreshold)
        {
            Dictionary<object,Dictionary<object,double>> mergedLinkages=new Dictionary<object, Dictionary<object, double>>();
            int processed = 0;
            int total = key2_key1_scores.Count;
            foreach (object key2 in key2_key1_scores.Keys)
            {
                if(this.WorkerStatusChangedEvent !=null && processed % 100==0)
                {
                    string msg = string.Format("Merging entity {0}/{1} links, processed {2} of {3} links",
                                               this.instruction.TableName1, this.instruction.TableName2, processed, total);
                    int pct = processed*100/total;
                    this.WorkerStatusChangedEvent(this._TicketID, this.GetType(), this.instruction,
                                                  this._WorkflowName, this._ComponentName, msg, pct);
                }

                Dictionary<object, double> toEntity1Scores = key2_key1_scores[key2];
                if (toEntity1Scores == null || toEntity1Scores.Count == 0)
                    continue;

                object key1 = null;
                double maxScore = 0;
                foreach (object key in toEntity1Scores.Keys)
                {
                    if (toEntity1Scores[key] > maxScore)
                    {
                        maxScore = toEntity1Scores[key];
                        key1 = key;
                    }
                }
                if (maxScore >= linkThreshold)
                {
                    toEntity1Scores=new Dictionary<object, double>();
                    toEntity1Scores.Add(key1, maxScore);
                    mergedLinkages.Add(key2, toEntity1Scores);
                }
                processed++;
            }
            return mergedLinkages;
        }
        #endregion

        #region test
        /// <summary>
        /// 
        /// </summary>
        /// <param name="entityID1"></param>
        /// <param name="entityID2"></param>
        /// <returns></returns>
        public DataTable TestEntityLinks(
            string entityID1, string entityID2)
        {
            TableColumn idCol1 = new TableColumn();
            if (string.IsNullOrEmpty(this.instruction.Sql1))
                idCol1 = DbUtil.GetTableColumn(DataProviderType.MSSQL, this.instruction.ConnStr1, this.instruction.SchemaName1, this.instruction.TableName1, this.instruction.PKField1);
            else
                idCol1 = DbUtil.GetTableColumn(DataProviderType.MSSQL, this.instruction.ConnStr1, this.instruction.Sql1, this.instruction.PKField1);
            TableColumn idCol2 = new TableColumn();
            if (string.IsNullOrEmpty(this.instruction.Sql1))
                idCol2 = DbUtil.GetTableColumn(DataProviderType.MSSQL, this.instruction.ConnStr2, this.instruction.SchemaName2, this.instruction.TableName2, this.instruction.PKField2);
            else
                idCol2 = DbUtil.GetTableColumn(DataProviderType.MSSQL, this.instruction.ConnStr2, this.instruction.Sql2, this.instruction.PKField2);

            string sql1 = string.Empty;
            if (string.IsNullOrEmpty(instruction.Sql1))
                sql1 = string.Format("select * from {0}.{1} where {2}={3}",
                    this.instruction.SchemaName1,
                    this.instruction.TableName1,
                    this.instruction.PKField1,
                    FieldDataType.ToMssqlDataValue(idCol1.DbType, entityID1));
            else
                sql1 = string.Format("select * from ({0}) t1 where t1.{1}={2}",
                    this.instruction.Sql1,
                    this.instruction.PKField1,
                    FieldDataType.ToMssqlDataValue(idCol1.DbType, entityID1));
            DataTable dt1 = DbUtil.GetTableDataBySQL(
                this.instruction.ConnStr1, "Dt1", sql1);
            Debug.Assert(dt1 != null && dt1.Rows.Count == 1);

            string sql2 = string.Empty;
            if (string.IsNullOrEmpty(instruction.Sql2))
                sql2 = string.Format("select * from {0}.{1} where {2}={3}",
                    this.instruction.SchemaName2,
                    this.instruction.TableName2,
                    this.instruction.PKField2,
                    FieldDataType.ToMssqlDataValue(idCol2.DbType, entityID2));
            else
                sql2 = string.Format("select * from ({0}) t1 where t1.{1}={2}",
                    this.instruction.Sql2,
                    this.instruction.PKField2,
                    FieldDataType.ToMssqlDataValue(idCol2.DbType, entityID2));
            DataTable dt2 = DbUtil.GetTableDataBySQL(this.instruction.ConnStr2, "DT2", sql2);
            Debug.Assert(dt2 != null && dt2.Rows.Count == 1);

            DataTable dt = new DataTable();
            dt.Columns.Add("FieldName", typeof(string));
            dt.Columns.Add("Field1", typeof(string));
            dt.Columns.Add("Field2", typeof(string));
            dt.Columns.Add("Score", typeof(double));
            Assembly assembly = this.GetType().Assembly;
            foreach (string fieldName in this.instruction.Field1_Field2_Mappings.Keys)
            {
                string field1 = fieldName;
                string field2 = this.instruction.Field1_Field2_Mappings[field1];
                object fieldValue1 = null;
                if (dt1.Columns.Contains(field1))
                    fieldValue1 = dt1.Rows[0][field1];
                object fieldValue2 = null;
                if (dt2.Columns.Contains(field2))
                    fieldValue2 = dt2.Rows[0][field2];
                double score = 0;
                if (this.instruction.FieldComparers != null && this.instruction.FieldComparers.ContainsKey(field1))
                {
                    Type comparerType = assembly.GetType(this.instruction.FieldComparers[field1]);
                    IAttributeComparer comparer = (IAttributeComparer)Activator.CreateInstance(comparerType);
                    double penalty = 0;
                    if (this.instruction.FieldPenalties != null && this.instruction.FieldPenalties.ContainsKey(field1))
                        penalty = this.instruction.FieldPenalties[field1];
                    score = comparer.Compare(fieldValue1, fieldValue2, penalty, null, null);
                    double weight = 1;
                    if (this.instruction.FieldWeights != null && this.instruction.FieldWeights.ContainsKey(field1))
                        weight = this.instruction.FieldWeights[field1];
                    score = score * weight;
                }
                DataRow dr = dt.NewRow();
                dr["FieldName"] = field1;
                dr["Field1"] = (fieldValue1 == null || fieldValue1 == DBNull.Value) ? string.Empty : fieldValue1.ToString();
                dr["Field2"] = (fieldValue2 == null || fieldValue2 == DBNull.Value) ? string.Empty : fieldValue2.ToString();
                dr["Score"] = score;
                dt.Rows.Add(dr);
            }

            return dt;
        }
        #endregion
    }
}
